package main

import (
	"container/list"
	"errors"
	"os/exec"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	log "github.com/Sirupsen/logrus"
)

func (this *JobHandler) CmdJobExecute(job JobConf) int {
	defer printStack()
	log.Infof("cmd job execute %+v", job)
	jobId := job.Get(CONF_JOB_ID)
	runJob := &RunningJob{
		jobConf: job,
	}
	this.runningJobs.Set(jobId, runJob)
	jobResult := this.executeJob(job)
	//等所有的日志都回写完成
	<-this.sendDone
	// 执行完成后删除job
	this.runningJobs.Delete(jobId)

	jobResult.SendIndex = atomic.AddInt32(&sendIndex, 1)
	msg := &Message{
		Type: MT_JOB_RESULT,
		Body: jobResult,
	}
	for {
		log.Debugf("submit job result: %+v", jobResult)
		this.ctx.FlushMessage(msg)
		err := this.waitLogResp(msg)
		if err != nil {
			log.Errorf("wait job result resp: %s", err.Error())
			time.Sleep(time.Second)
			continue
		}
		break
	}
	return 0
}

func (this *JobHandler) KillJobExecute(job JobConf) {
	log.Infof("kill job execute %+v", job)
	jobResult := NewJobResult()
	jobResult.Id = job.Get(CONF_JOB_ID)
	jobResult.Status = JOB_STATUS_KILLED
	jobResult.JobConf = job
	err := this.KillJob(job)
	if err != nil {
		jobResult.Status = JOB_STATUS_KILL_FAILED
		jobResult.ErrDesc = err.Error()
	}
	jobId := job.Get(CONF_JOB_ID)
	this.runningJobs.Delete(jobId)
	msg := &Message{
		Type: MT_JOB_RESULT,
		Body: jobResult,
	}
	err = this.ctx.FlushMessage(msg)
	if err != nil {
		log.Errorf("write jobresult error: %s", err.Error())
	}
}

func (this *JobHandler) KillJob(job JobConf) error {
	defer printStack()
	jobId := job.Get(CONF_JOB_ID)
	runJob := this.runningJobs.Get(jobId)
	if runJob != nil {
		log.Infof("kill job %+v", job)
		err := runJob.cmd.Process.Kill()
		if err != nil {
			log.Errorf("cmd process kill error: %s", err.Error())
		}
		return err
	}
	return nil
}

type ErrList []error

func (l ErrList) Error() string {
	var es []string
	for _, e := range l {
		es = append(es, e.Error())
	}
	return strings.Join(es, "\n")
}

func (l ErrList) Errors() []error {
	return l
}

func (this *JobHandler) KillAllJob() error {
	var errors []error
	for runJob := range this.runningJobs.Iter() {
		err := this.KillJob(runJob.jobConf)
		if err != nil {
			errors = append(errors, err)
		}
	}
	if len(errors) != 0 {
		return ErrList(errors)
	}
	return nil
}

func (this *JobHandler) executeJob(jobConf JobConf) *JobResult {
	cmdArray, err := this.parseCmd(jobConf)
	log.Debugf("cmd array length: %d, %+v", len(cmdArray), cmdArray)
	if err != nil {
		result := &JobResult{}
		result.Id = jobConf.Get(CONF_JOB_ID)
		result.Status = JOB_STATUS_FAILED
		result.JobConf = jobConf
		result.ErrDesc = err.Error()
		return result
	}
	ret, desc := this.execProcess(jobConf, cmdArray)
	result := &JobResult{}
	result.Id = jobConf.Get(CONF_JOB_ID)
	result.Status = JOB_STATUS_FAILED
	result.JobConf = jobConf
	if ret == 0 {
		result.Status = JOB_STATUS_SUCCEED
	} else {
		result.Status = JOB_STATUS_FAILED
	}
	result.ErrDesc = desc
	return result
}

/*
解析cmdArray为可执行的命令格式
*/
func (this *JobHandler) execProcess(jobConf JobConf, cmdArray []string) (int, string) {
	cdIndex := 0
	otherCmdCount := 0
	suBool := false
	finalArrayIndex := 0
	if strings.Index(cmdArray[0], "su ") == 0 {
		otherCmdCount++
		cdIndex = 1
		suBool = true
		finalArrayIndex = finalArrayIndex + 3
	}
	if strings.Index(cmdArray[cdIndex], "cd ") == 0 {
		otherCmdCount++
		pathStr := strings.Split(cmdArray[cdIndex], " ")[1]
		var finalCmdArray []string
		if suBool {
			if otherCmdCount+1 < len(cmdArray) {
				finalCmdArray = make([]string, len(cmdArray)+2)
				finalCmdArray[0] = "su"
				finalCmdArray[1] = "-"
				//切换的用户名
				finalCmdArray[2] = strings.Split(cmdArray[0], " ")[1]
				//文件名
				if strings.LastIndex(pathStr, "/") == len(pathStr)-1 {
					finalCmdArray[3] = pathStr + cmdArray[otherCmdCount]
				} else {
					finalCmdArray[3] = pathStr + "/" + cmdArray[otherCmdCount]
				}

				finalCmdArray[4] = "--"
				i := 5
				j := otherCmdCount + 1
				for j < len(cmdArray) {
					finalCmdArray[i] = cmdArray[j]
					i++
					j++
				}
			} else {
				finalCmdArray = make([]string, len(cmdArray)+1)
				finalCmdArray[0] = "su"
				finalCmdArray[1] = "-"
				//切换的用户名
				finalCmdArray[2] = strings.Split(cmdArray[0], " ")[1]
				//文件名
				if strings.LastIndex(pathStr, "/") == len(pathStr)-1 {
					finalCmdArray[3] = pathStr + cmdArray[otherCmdCount]
				} else {
					finalCmdArray[3] = pathStr + "/" + cmdArray[otherCmdCount]
				}
			}
		} else {
			finalCmdArray = make([]string, len(cmdArray)-otherCmdCount+finalArrayIndex+4)
			finalCmdArray[0] = "su"
			finalCmdArray[1] = "-"
			finalCmdArray[2] = "root"
			if strings.LastIndex(pathStr, "/") == len(pathStr)-1 {
				finalCmdArray[3] = pathStr + cmdArray[otherCmdCount]
			} else {
				finalCmdArray[3] = pathStr + "/" + cmdArray[otherCmdCount]

			}
			finalCmdArray[4] = "--"
			i := finalArrayIndex + 5
			j := otherCmdCount + 1
			for j < len(cmdArray) {
				finalCmdArray[i] = cmdArray[j]
				i++
				j++
			}
		}
		return this.execProcess0(jobConf, finalCmdArray, nil, pathStr)
	} else {
		var finalCmdArray []string
		if suBool {
			if otherCmdCount+1 < len(cmdArray) {
				finalCmdArray = make([]string, len(cmdArray)+3)
				finalCmdArray[0] = "su"
				finalCmdArray[1] = "-"
				//切换的用户名
				finalCmdArray[2] = strings.Split(cmdArray[0], " ")[1]
				//文件名
				finalCmdArray[3] = cmdArray[otherCmdCount]
				finalCmdArray[4] = "--"
				i := 5
				j := otherCmdCount + 1
				for j < len(cmdArray) {
					finalCmdArray[i] = cmdArray[j]
					i++
					j++
				}
			} else {
				finalCmdArray = make([]string, len(cmdArray)+2)
				finalCmdArray[0] = "su"
				finalCmdArray[1] = "-"
				//切换的用户名
				finalCmdArray[2] = strings.Split(cmdArray[0], " ")[1]
				//文件名
				finalCmdArray[3] = cmdArray[otherCmdCount]
			}
		} else {
			finalCmdArray = make([]string, len(cmdArray)-otherCmdCount+finalArrayIndex+4)
			finalCmdArray[0] = "su"
			finalCmdArray[1] = "-"
			finalCmdArray[2] = "root"
			finalCmdArray[3] = cmdArray[otherCmdCount]
			finalCmdArray[4] = "--"
			i := finalArrayIndex + 5
			j := otherCmdCount + 1
			for j < len(cmdArray) {
				finalCmdArray[i] = cmdArray[j]
				i++
				j++
			}
		}
		return this.execProcess0(jobConf, finalCmdArray, nil, "")
	}
}

/*
真正的执行命令
*/
func (this *JobHandler) execProcess0(jobConf JobConf, cmdExecArray []string, env []string, dir string) (int, string) {
	log.Infof("exec process cmdArray: %+v, env: %+v, dir: %s", cmdExecArray, env, dir)
	defer func() {
		//this.done <- 1
		log.Info("exec job done!")
		close(this.done)
	}()
	cmd := exec.Command(cmdExecArray[0], cmdExecArray[1:]...)
	if len(env) != 0 {
		cmd.Env = env
	}
	if dir != "" {
		cmd.Dir = dir
	}
	runJob := this.runningJobs.Get(jobConf.Get(CONF_JOB_ID))
	if runJob != nil {
		runJob.cmd = cmd
	}
	output := new(Buffer)
	cmd.Stdout = output
	cmd.Stderr = output
	//go this.writeLog(jobConf, output)
	lst := list.New()
	go this.cutLogBuf(jobConf, output, lst)
	go this.sendLog(lst)
	err := cmd.Start()
	if err != nil {
		log.Errorf("start error %s: exec process cmdArray: %+v, env: %+v, dir: %s", err.Error(), cmdExecArray, env, dir)
		output.WriteString(err.Error())
		return 1, err.Error()
	}
	err = cmd.Wait()
	if err != nil {
		log.Errorf("wait error %s: exec process cmdArray: %+v, env: %+v, dir: %s", err.Error(), cmdExecArray, env, dir)
		output.WriteString(err.Error())
		return 1, err.Error()
	}
	if cmd.ProcessState.Success() {
		return 0, ""
	} else {
		return JOB_STATUS_FAILED, ""
	}
}

func (this *JobHandler) cutLogBuf(jobConf JobConf, buf *Buffer, lst *list.List) {
	//var sendIndex int
	ticker := time.NewTicker(time.Second * 5)
	for {
		select {
		case <-this.done:
			for data := buf.Next(Conf.MaxPerLogSize); len(data) > 0; data = buf.Next(Conf.MaxPerLogSize) {
				result := NewJobResult()
				result.Id = jobConf.Get(CONF_JOB_ID)
				result.Status = JOB_STATUS_RUNNING
				result.JobConf = jobConf
				result.ErrDesc = string(data)
				log.Debugf("generate JobResult [%d]: %+v", sendIndex, result)
				message := &Message{
					Type: MT_LOG_SUBMIT,
					Body: result,
				}
				lst.PushBack(message)
			}
			close(this.logDone)
			log.Infof("cut log buf done!")
			return
		// 任务已经执行完成
		case <-ticker.C:
			// 每隔一段时间将buf中的数据分割生成一个JobResult对象，放入队列
			data := buf.Next(Conf.MaxPerLogSize)
			if len(data) != 0 {
				result := NewJobResult()
				result.Id = jobConf.Get(CONF_JOB_ID)
				result.Status = JOB_STATUS_RUNNING
				result.JobConf = jobConf
				result.ErrDesc = string(data)
				log.Debugf("generate JobResult [%d]: %+v", sendIndex, result)
				message := &Message{
					Type: MT_LOG_SUBMIT,
					Body: result,
				}
				lst.PushBack(message)
			}
		}
	}
}

/*
定时回写日志
*/
func (this *JobHandler) writeLog(jobConf JobConf, buf *Buffer) {
	ticker := time.NewTicker(Conf.SubmitLogInterval.Duration)
	ticker1 := time.NewTicker(time.Second)
	for {
		select {
		case <-this.done:
			length := buf.Len()
			// 确保剩下的日志发送完成，才能退出
			for length != 0 {
				this.sendOneMessage(jobConf, buf, buf.Len())
				time.Sleep(time.Second)
				length = buf.Len()
			}
			log.Info("submit log done")
			this.logDone <- 1
			return
		case <-ticker.C:
			if buf.Len() != 0 {
				this.sendOneMessage(jobConf, buf, buf.Len())
			}
		case <-ticker1.C:
			if buf.Len() > Conf.MaxPerLogSize {
				this.sendOneMessage(jobConf, buf, Conf.MaxPerLogSize)
			}
		}
	}
}

/*
定时回写日志
*/
func (this *JobHandler) sendLog(lst *list.List) {
	ticker := time.NewTicker(Conf.SubmitLogInterval.Duration)
	for {
		select {
		//日志已经生成完成
		case <-this.logDone:
			for e := lst.Front(); e != nil; e = lst.Front() {
				message := e.Value.(*Message)
				log.Debugf("submit log: %+v", message.Body)
				err := this.ctx.FlushMessage(message)
				if err != nil {
					log.Errorf("flush message: %s", err.Error())
					time.Sleep(time.Second * 5)
					continue
				}
				err = this.waitLogResp(message)
				if err == nil {
					lst.Remove(e)
					if job, ok := message.Body.(*JobResult); ok {
						lastSendSuccessIndex = job.SendIndex
					}
				} else {
					log.Errorf("receive log resp error: %+s", err.Error())
					time.Sleep(time.Second * 5)
					continue
				}
			}
			close(this.sendDone)
			log.Info("submit log done!")
			return
		//定时发送日志消息
		case <-ticker.C:
			e := lst.Front()
			if e == nil {
				continue
			}
			message := e.Value.(*Message)
			log.Debugf("submit log: %+v", message.Body)
			err := this.ctx.FlushMessage(message)
			if err != nil {
				log.Errorf("flush message: %s", err.Error())
				continue
			}

			err = this.waitLogResp(message)
			if err == nil {
				lst.Remove(e)
				if job, ok := message.Body.(*JobResult); ok {
					lastSendSuccessIndex = job.SendIndex
				}
			} else {
				log.Errorf("receive log resp error: %+s", err.Error())
			}
		}
	}
}

var oneMsgCount int32 = 1

func (this *JobHandler) sendOneMessage(jobConf JobConf, buf *Buffer, size int) {
	log.Infof("send one message %d", oneMsgCount)
	atomic.AddInt32(&oneMsgCount, 1)
	data, _ := buf.Peek(size)
	result := &JobResult{
		Id:      jobConf.Get(CONF_JOB_ID),
		Status:  JOB_STATUS_RUNNING,
		JobConf: jobConf,
	}
	result.ErrDesc = string(data)
	message := &Message{
		Type: MT_LOG_SUBMIT,
		Body: result,
	}
	log.Infof("submit log: %s", string(data))
	err := this.ctx.FlushMessage(message)
	if err != nil {
		log.Errorf("flush message: %s", err.Error())
	}

	err = this.waitLogResp(message)
	if err == nil {
		// move forword buf
		buf.Forward(size)
	} else {
		log.Errorf("receive log resp error: %+s", err.Error())
	}
}

var respCount int32 = 1

func (this *JobHandler) waitLogResp(message *Message) error {
	select {
	case success := <-this.logRespChan:
		if success {
			log.Infof("receive log response for %d", respCount)
			atomic.AddInt32(&respCount, 1)
		} else {
			return errors.New("receive log response failed")
		}
	case <-time.After(Conf.LogResponseTimeout.Duration):
		return errors.New("receive log response timeout")
	}
	return nil
}

/*
将传过来的参数解析为包含执行用户，目录等信息的数组
*/
func (this *JobHandler) parseCmd(jobConf JobConf) ([]string, error) {
	var argc int
	var cmdArray = make([]string, 0, 5)
	argc, err := strconv.Atoi(jobConf.Get("job.command.arg.count"))
	if err != nil {
		argc = -1
	}
	if argc < 0 {
		return nil, errors.New("job command arg count < 0")
	}
	user := jobConf.Get("job.command.user")
	if user != "" {
		cmdArray = append(cmdArray, "su "+user)
	}
	path := jobConf.Get("job.command.path")
	if path != "" {
		cmdArray = append(cmdArray, "cd "+path)
	}
	filename := jobConf.Get("job.command.name")
	if filename == "" {
		return nil, errors.New("job command name is empty")
	}
	cmdArray = append(cmdArray, filename)
	for i := 0; i < argc; i++ {
		arg := jobConf.Get("job.command.arg." + strconv.Itoa(i))
		if arg == "" {
			return nil, errors.New("job command arg is empty")
		}
		cmdArray = append(cmdArray, arg)
	}
	return cmdArray, nil
}
